package com.yc.bigdata.flink.demo.util;

import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.api.RpcClient;
import org.apache.flume.api.RpcClientFactory;
import org.apache.flume.event.EventBuilder;

import java.nio.charset.Charset;

/**
 * <p></p>
 *
 * @author: YuanChilde
 * @date: 2020-02-14 14:04
 * @version: 1.0
 * Modification History:
 * Date    Author      Version     Description
 * -----------------------------------------------------------------
 * 2020-02-14 14:04    YuanChilde     1.0        新增
 */
public class FlumeAvroSendData {
    public static void main(String[] args) {
        MyRpcClientFacade client = new MyRpcClientFacade();
        client.init("dev.dongbaosoft.com", 41414);
        String sampleData = "Hello Flume!";
        for (int i = 0; i < 5; i++) {
            client.sendDataToFlume(sampleData+" " + i);
        }
        System.out.println("输出完毕");
        client.cleanUp();
    }
    static class MyRpcClientFacade {
        private RpcClient client;
        private String hostname;
        private int port;

        public void init(String hostname, int port) {
            // Setup the RPC connection
            this.hostname = hostname;
            this.port = port;
            this.client = RpcClientFactory.getDefaultInstance(hostname, port);
        }

        public void sendDataToFlume(String data) {
            Event event = EventBuilder.withBody(data, Charset.forName("UTF-8"));
            try {
                client.append(event);
            } catch (EventDeliveryException e) {
                client.close();
                client = null;
                client = RpcClientFactory.getDefaultInstance(hostname, port);
            }
        }

        public void cleanUp() {
            client.close();
        }
    }
}

